package com.csdn.eval.spring.mqtt.core;


import com.csdn.eval.spring.mqtt.constants.Common;
import com.csdn.eval.spring.mqtt.constants.TopicConst;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageHeaders;

import java.util.concurrent.atomic.AtomicInteger;

public class MqttGatewayPublish<T> {

    private static final Logger log = LoggerFactory.getLogger(MqttGatewayPublish.class);

    private static final int DEFAULT_QOS = 0;
    public static final int DEFAULT_RETRY_COUNT = 2;
    public static final int DEFAULT_RETRY_TIMEOUT = 3000;

    @Autowired(required = false)
    private IMqttMessageGateway messageGateway;

    public void publish(String topic, int qos, T request) {
        try {
            log.debug("send topic: {}, payload: {}", topic, request.toString());
            byte[] payload = Common.getObjectMapper().writeValueAsBytes(request);
            messageGateway.publish(topic, payload, qos);
        } catch (JsonProcessingException e) {
            log.error("Failed to publish the message. {}", request.toString());
            e.printStackTrace();
        }
    }

    public void publish(String topic, T request, int publishCount) {
        AtomicInteger time = new AtomicInteger(0);
        while (time.getAndIncrement() < publishCount) {
            this.publish(topic, DEFAULT_QOS, request);
        }
    }

    public void publish(String topic, T request) {
        this.publish(topic, DEFAULT_QOS, request);
    }

    public void publishReply(T response, MessageHeaders headers) {
        this.publish(headers.get(MqttHeaders.RECEIVED_TOPIC) + TopicConst._REPLY_SUF, 2, response);
    }

    public void publishReply(T response, int qos, String topic) {
        this.publish(topic, qos, response);
    }

    public void publishReply(Object object, String topic) {
        try {
            byte[] payload = Common.getObjectMapper().writeValueAsBytes(object);
            messageGateway.publish(topic, payload, DEFAULT_QOS);
        } catch (JsonProcessingException e) {
            log.error("Failed to publish the message", e);
            e.printStackTrace();
        }
    }


}